package com.pk.flink.apps.datasource;


import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.ParallelSourceFunction;
import org.apache.flink.util.NumberSequenceIterator;

import java.util.Arrays;

/**
 * 使用Flink 实现单词计数
 * @author pk
 */
public class FlinkDataSourceApp {
    public static void main(String[] args) throws Exception {
        //获取上下文环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //DataStreamSource<Integer> integerDataStreamSource = env.fromElements(1, 2, 3, 4);
        DataStreamSource<Integer> integerDataStreamSource = env.fromCollection(Arrays.asList(1, 2, 3, 4));
        System.out.println(integerDataStreamSource.getParallelism());

        SingleOutputStreamOperator<Integer> map = integerDataStreamSource.map(x -> x * 10).setParallelism(8);
        System.out.println(map.getParallelism());

        map.print();

        env.execute();

//        DataStreamSource<Long> longDataStreamSource = env.fromParallelCollection(new NumberSequenceIterator(1, 10),Long.TYPE);
//        System.out.println(longDataStreamSource.getParallelism());
//
//        SingleOutputStreamOperator<Long> longSingleOutputStreamOperator = longDataStreamSource.map(x -> x + 100).setParallelism((3));
//        System.out.println(longSingleOutputStreamOperator.getParallelism());
//
//        longSingleOutputStreamOperator.print();
//        env.execute();
    }
}
